package com.geek.activemq.jms;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;

import javax.jms.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @program active-mq
 * @description: activemq
 * @author: chanzhugang
 * @create: 2021/07/28 13:52
 */
public class ActiveMQDemo {

    public static void main(String[] args) {

      //  Destination destination = new ActiveMQTopic("test.topic");

        ActiveMQQueue destination = new ActiveMQQueue("test.queue");

        testDestination(destination);

    }

    private static void testDestination(Destination destination) {
        // 创建连接、会话
        try {
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61617");
            Connection conn = factory.createConnection();
            conn.start();
            Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 创建消费者
            MessageConsumer consumer = session.createConsumer(destination);
            AtomicInteger count = new AtomicInteger(0);
            // 创建监听器绑定到消费者
            MessageListener listener = new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    try {
                        System.out.println(count.incrementAndGet() +"====> receive from " + destination.toString() + ":" + message);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            };
            consumer.setMessageListener(listener);

            // 创建生产者
            MessageProducer producer = session.createProducer(destination);
            int index = 0;
            while (index++ < 100) {
                TextMessage textMessage = session.createTextMessage(index + " message");
                producer.send(textMessage);
            }

            Thread.sleep(5000);
            session.close();
            conn.close();
        } catch (Exception e) {
            e.printStackTrace();
        }



    }
}